package com.yangchao.webflux.sink;

import com.yangchao.webflux.projectreactor.MonoTest;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

public class SinkTest {


    static Logger logger = org.slf4j.LoggerFactory.getLogger(MonoTest.class);

    @Test
    @DisplayName("Sink test01")
    public void test01() throws InterruptedException {
        Flux<String> flux = Flux.generate(
                () -> 0,
                (state, sink) -> {
                    sink.next("3 x " + state + " = " + 3 * state);
                    if (state == 10) {
                        sink.complete();
                    }
                    return state + 1;
                });
        flux.subscribe(System.out::println);
    }

    @Test
    @DisplayName("Sink test02")
    public void test02() throws InterruptedException {
        Flux<String> flux = Flux.generate(
                AtomicLong::new,
                (state, sink) -> {
                    long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3*i);
                    if (i == 10) sink.complete();
                    return state;
                });
        flux.subscribe(System.out::println);
    }

    @Test
    @DisplayName("Sink test03")
    public void test03() throws InterruptedException {
        Flux<String> bridge = Flux.create(new Consumer<FluxSink<String>>() {
            @Override
            public void accept(FluxSink<String> stringFluxSink) {
                logger.info("{}", stringFluxSink);
                stringFluxSink.next("1");
                stringFluxSink.next("2");
                stringFluxSink.next("3");
                stringFluxSink.complete();
            }
        });
        bridge.subscribe(e->{
            logger.info("{}", e);
        });
    }
}
